package springbootreactor;

import java.util.concurrent.CountDownLatch;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.IntegrationTest;
import org.springframework.boot.test.OutputCapture;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import reactor.core.Reactor;
import reactor.event.Event;
import reactor.function.Consumer;
import reactor.function.Function;

import static org.hamcrest.CoreMatchers.containsString;
import static reactor.event.selector.Selectors.$;
import static org.junit.Assert.*;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes=ReactorApplication.class)
@IntegrationTest
public class ReactorTest {
    private static Log log=LogFactory.getLog(ReactorTest.class);

    @Rule
    public OutputCapture output=new OutputCapture();
    @Autowired
    private Reactor reactor;
    @Autowired
    private SayhitConsumerEventHandler sayhitConsumerEventHandler;


    @Autowired
    private SayhitReplyHandler sayhitReplyHandler;

    protected Reactor notify(String key,Event<String> ev)throws Exception{
        final CountDownLatch latch=new CountDownLatch(1);

        Reactor r=reactor.notify(key, ev, new Consumer<Event<String>>() {
            @Override
            public void accept(Event<String> evt) {
                log.info(evt.getKey()+" complete: [data:"+evt.getData()+",replyTo:"+evt.getReplyTo()+"]");
                latch.countDown();
            }
        });
        log.info(ev.getKey()+" notify");
        latch.await();
        System.out.println(r.getId());
        return r;
    }

    @Test
    public void testSayhitConsumerEvent()throws Exception{
        reactor.on($("testSayhitConsumerEvent"), sayhitConsumerEventHandler);
        notify("testSayhitConsumerEvent", Event.wrap("blog.163.com/mr_zyf"));
        //assertThat(output.toString(),containsString("blog.163.com/mr_zyf"));
        System.out.println("========================="+output.toString());
    }

    @Test
    public void testSayhit()throws Exception{
        notify("testSayhit", Event.wrap("yoyo"));
        assertThat(output.toString(),containsString("yoyo"));
    }

    @Test
    public void testSayhitEvent()throws Exception{
        notify("testSayhitEvent", Event.wrap("judy"));
        assertThat(output.toString(),containsString("judy"));
    }
    @Test
    public void testSayhitReply1()throws Exception{
        notify("testSayhitReply1", Event.wrap("tonyzhao"));
        assertThat(output.toString(),containsString("tonyzhao"));
    }
    @Test
    public void testSayhitReply2()throws Exception{
        notify("testSayhitReply2", Event.wrap("tony100").setReplyTo("testSayhitReplyResult2"));
        assertThat(output.toString(),containsString("tony100"));
    }
    @Test
    public void testSayhitReply3() throws Exception {
        notify("testSayhitReply3", Event.wrap("tony100.net"));
        assertThat(output.toString(), containsString("tony100.net"));
    }

    @Test
    public void testReactorReceive() throws Exception {
        reactor.on($("testReactorReceiveResult"),
                new Consumer<Event<String>>() {
                    @Override
                    public void accept(Event<String> event) {
                        // TODO Auto-generated method stub
                        log.info("testReactorReceiveResult accept:"
                                + event.getData());
                    }
                });
        reactor.receive($("testReactorReceive"),
                new Function<Event<String>, String>() {

                    @Override
                    public String apply(Event<String> event) {
                        // TODO Auto-generated method stub
                        log.info("testReactorReceive apply:" + event.getData());
                        return event.getData() + " hello world!";
                    }
                });
        notify("testReactorReceive",
                Event.wrap("DIY Seactor!").setReplyTo(
                        "testReactorReceiveResult"));

        //目测一下，notify 中的第一个参数，是一个处理方，将这个
        // Event交给第一个参数代表的类进行处理，ReplyTo 语义是回答
        // 如果设置了，就会再到ReplyTo里面再继续执行，如果没设置不报错，
        // 设置了没有也不报错（整个reactor就不报错）
        // 可以手动指定Consumer,当然是可被Spring扫到，请见SayhitConsumerEventHandler类
        // 也可以通过注解的形式，使用@Consumer 注解，表示是一个Consumer,这样就不需要实现
        // Consumer接口了，一个被@Consumer表示的类，里面可能有很多的处理方法，每个处理的方法需要用
        // @Selector(value="testSayhit"),其中的value 就是Selector的名字

    }


    @Test
    public void testlinyang(){
        reactor.notify("linyang", Event.wrap("linyang"), new Consumer<Event<String>>() {
            @Override
            public void accept(Event<String> stringEvent) {

            }
        });
    }
}